package com.platform.apex.output;

import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;

import javax.validation.constraints.Min;

import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.lib.db.AbstractStoreOutputOperator;
import com.datatorrent.netlet.util.DTThrowable;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.platform.apex.util.ConvertUtil;

public abstract class AbstractAppElasticSearchOutputOperator<T, S extends ElasticAppSearchConnectable>
		extends AbstractStoreOutputOperator<T, S> {
	
	protected static final int DEFAULT_BATCH_SIZE = 1000;
	
	@Min(1)
	protected int batchSize = DEFAULT_BATCH_SIZE;
	protected transient Queue<T> tupleBatch;
	protected Set<String> cacheIndex = new HashSet<>();
	private static ObjectMapper objectMapper = new ObjectMapper();
	

	/**
	 * Initialize transient fields such as {@code tupleBatch}
	 *
	 * @see com.datatorrent.lib.db.AbstractStoreOutputOperator#setup(com.datatorrent.api.Context.OperatorContext)
	 */
	@Override
	public void setup(OperatorContext context) {
		super.setup(context);
		tupleBatch = new ArrayBlockingQueue<T>(batchSize);
	}

	/**
	 * Adds tuple to the queue. Calls {@link #processBatch()} if queue is full
	 *
	 * @see com.datatorrent.lib.db.AbstractStoreOutputOperator#processTuple(java.lang.Object)
	 */
	public void processTuple(T tuple) {
		tupleBatch.add(tuple);
		try {
			System.out.println(objectMapper.writeValueAsString(tuple));
		} catch (JsonProcessingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		if (tupleBatch.size() >= batchSize) {
			processBatch();
		}
	}

	 @Override
	 public void beginWindow(long windowId) {
		 super.beginWindow(windowId);
	 }
	  
	/**
	 * Flush the batch queue at end window
	 */
	@Override
	public void endWindow() {
		super.endWindow();
		processBatch();
	}

	/**
	 * This will flush all the tuples from queue to ElasticSearch. It uses
	 * BulkRequestBuilder API for sending batch.
	 */
	private void processBatch() {

//		store.client.admin().indices().i
		boolean isAdded = false;
		BulkRequestBuilder bulkRequestBuilder = store.client.prepareBulk();
		while (!tupleBatch.isEmpty()) {
			T tuple = tupleBatch.remove();
			
			Map<String,Object> dempMap = (Map<String,Object>)tuple;
			
			String demo=null;
			try {
				demo = objectMapper.writeValueAsString(dempMap);
			} catch (JsonProcessingException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			if(demo!=null){
				boolean indexExist = checkIndexExists(getIndexName(tuple));
				if (!indexExist) {
					createIndexType(getIndexName(tuple), getType(tuple),demo,tuple);
				}
			}
			
			IndexRequestBuilder indexRequestBuilder = getIndexRequestBuilder(tuple);
			System.out.println("send success");
			bulkRequestBuilder.add(indexRequestBuilder);
			isAdded = true;
		}

		if (isAdded) {
			BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
			if (bulkResponse.hasFailures()) {
				System.err.println("bulkResponse error-->" + bulkResponse.buildFailureMessage());
				DTThrowable.rethrow(new Exception(bulkResponse.buildFailureMessage()));
			}
		}
	}

	/**
	 * Create {@link IndexRequestBuilder} for this tuple. It calls
	 * {@link #getId(T)}, {@link #getIndexName(T)}, {@link #getType(T)}.
	 *
	 * @param tuple
	 * @return
	 */
	protected IndexRequestBuilder getIndexRequestBuilder(T tuple) {
		IndexRequestBuilder indexRequestBuilder = store.client.prepareIndex(getIndexName(tuple), getType(tuple));;
		String id = getId(tuple);
		if (id != null) {
			indexRequestBuilder.setId(id);
		}
		indexRequestBuilder.setType(getType(tuple));
		return setSource(indexRequestBuilder, tuple);
	}

	/**
	 * User actions required for format conversions from tuple to one of the
	 * source types supported by IndexRequest.
	 * 
	 * @param tuple
	 * @return
	 */
	protected abstract IndexRequestBuilder setSource(IndexRequestBuilder indexRequestBuilder, T tuple);

	/**
	 * Determine id for the given tuple.<br>
	 * If tuples do not have any field mapping to unique id then this function
	 * may return null. In this case ElasticSearch will add auto-generated _id
	 * field to the document. ElasticSearch will treat every tuple as fresh
	 * record. If user requires {@code ProcessingMode.EXACTLY_ONCE} support;
	 * then user must implement {@link #getId(Object)} and make sure that
	 * distinct value is returned for each record.
	 *
	 * @param tuple
	 * @return
	 */
	public abstract String getId(T tuple);

	/**
	 * Decides indexName under which this tuple gets stored. Depending on
	 * scenario, user may decide index all tuples under same indexName (In this
	 * case user can return constant value for indexName). Or user may have
	 * custom logic for deriving indexName
	 * 
	 * @param tuple
	 * @return
	 */
	protected abstract String getIndexName(T tuple);

	/**
	 * Decides type for given tuple. Depending on scenario, user may decide
	 * index all tuples under same type (In this case user can return constant
	 * value for type). Or user may have custom logic for deriving type.
	 * 
	 * @param tuple
	 * @return
	 */
	protected abstract String getType(T tuple);

	/**
	 * @return the batchSize
	 */
	public int getBatchSize() {
		return batchSize;
	}

	/**
	 * @param batchSize
	 *            the batchSize to set
	 */
	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}

	
	private boolean checkIndexExists(String indexName) {
		if (cacheIndex.contains(indexName)) {
//			System.err.println("contains");
			return true;
		}
//		else {
//			System.err.println("XXXXXXXXcontains");
//		}
		IndicesExistsResponse response = store.client.admin().indices().prepareExists(indexName).execute().actionGet();
		if (response.isExists()) {
			cacheIndex.add(indexName);
		}
		return response.isExists();
	}

	public void createIndexType(String index, String type,String demo,T tuple) {
		Settings settings = Settings.settingsBuilder().put("cluster.name", getStore().getClusterName()).put("client.transport.sniff", true).build();
		try {
			store.client.admin().indices().prepareCreate(index).setSettings(settings).addMapping(type, getMapping(type,demo)).execute().actionGet();
			cacheIndex.add(getIndexName(tuple));
		} catch (Exception e) {
			System.err.println("createIndexType error ---> " + e.getMessage());
		}
	}
	
	private XContentBuilder getMapping(String type,String demo) throws Exception {
//		String[] cols = {"0","1","2","mg","mpg","5","6","SALES_DIST","8","9","a","b","WBS_ELEMT","d","e","f","g","h","i","j","k","l","m","n","o","p","q","r","s","t","u","v","w","x","y","z","a0","b0","c0","d0","e0","SALESORG","g0","h0","i0","j0","k0","l0","m0","GROSS_WGT","o0","p0","q0","r0","s0","t0","u0","v0","w0","x0","CML_CF_QTY","CML_CD_QTY","a1","b1","c1","d1","CML_OR_QTY","SUBTOTAL_1","SUBTOTAL_2","SUBTOTAL_3","SUBTOTAL_4","SUBTOTAL_5","SUBTOTAL_6","MIN_DL_QTY","m1","REQDEL_QTY","o1","p1","q1","r1","s1","t1","u1","TAX_VALUE","NET_PRICE","NET_VALUE","NET_WT_AP","z1","a2","b2","c2","d2","e2","f2","g2","h2","i2","DIVISION","k2","l2","m2","n2","UPPR_BND","DENOMINTR","NUMERATOR","DENOMINTRZ","NUMERATORZ","LOWR_BND","u2","v2","w2","x2","VOLUME_AP","z2","COST","b3","TARGET_QTY","TARG_VALUE","e3","f3","g3","h3","i3","j3","k3","l3","m3","n3","o3","p3","q3","r3","s3","t3","u3","v3","w3","x3","y3","z3","a4","b4","c4","ACCNT_ASGN","e4","f4","g4","h4","i4","j4","k4","l4","m4","n4","o4","p4","q4","BIC_ZSUBTA_9S","s4","t4","u4","v4","w4","x4","y4","z4","a5","BIC_ZSUBTA_7S","c5","d5","e5","f5","BIC_ZFV_DC","h5","mg1","mpg2","mg3","mpg3","salesORGGroup","salesDistGroup","hfmEntity","GROSS_WGT_ed","CML_CF_QTY_ed","CML_CD_QTY_ed","CML_OR_QTY_ed","SUBTOTAL_1_ed","SUBTOTAL_2_ed","SUBTOTAL_3_ed","SUBTOTAL_4_ed","SUBTOTAL_5_ed","SUBTOTAL_6_ed","MIN_DL_QTY_ed","REQDEL_QTY_ed","TAX_VALUE_ed","NET_PRICE_ed","NET_VALUE_ed","NET_WT_AP_ed","UPPR_BND_ed","DENOMINTR_ed","NUMERATOR_ed","DENOMINTRZ_ed","NUMERATORZ_ed","LOWR_BND_ed","VOLUME_AP_ed","COST_ed","TARGET_QTY_ed","TARG_VALUE_ed","BIC_ZSUBTA_9S_ed","BIC_ZSUBTA_7S_ed","BIC_ZFV_DC_ed","activity","afterSplitMG","afterSplitMPG"};
//	    String[] numberCols = {"GROSS_WGT","CML_CF_QTY","CML_CD_QTY","CML_OR_QTY","SUBTOTAL_1","SUBTOTAL_2","SUBTOTAL_3","SUBTOTAL_4","SUBTOTAL_5","SUBTOTAL_6","MIN_DL_QTY","REQDEL_QTY","TAX_VALUE","NET_PRICE","NET_VALUE","NET_WT_AP","UPPR_BND","DENOMINTR","NUMERATOR","DENOMINTRZ","NUMERATORZ","LOWR_BND","VOLUME_AP","COST","TARGET_QTY","TARG_VALUE","BIC_ZSUBTA_9S","BIC_ZSUBTA_7S","BIC_ZFV_DC"};	
//	    List<String> numberList = Arrays.asList(numberCols);
//	    String demo = "{\"DateTime\":123456789012,\"AllMsg\":\"wee\",\"TransactionID\":\"232\",\"ClientReq\":0.0,\"SihProcessTime\":0.0,\"ClientRes\":0.0,\"SvrRes\":0.0,\"ClientResAvMsgLen\":0.0,\"AppProcessTime\":0.0,\"SvrResAvMsgLen\":0.0,\"AppName\":\"TRR\",\"AppResponseMaxTime\":0.0,\"OtherSide\":\"MCSS_COMM01\",\"ClientReqAvMsgLen\":0.0,\"SvrReqAvMsgLen\":0.0,\"SvrReq\":0.0,\"SvrReqTps\":0.0,\"ClientResTps\":0.0,\"CoreProcessTime\":0.0,\"CoreProcessMaxTime\":0.0,\"SihProcessMaxTime\":0.0,\"Discard\":0.0,\"AppResponseTime\":0.0,\"MsgType\":\"No message1\",\"ClientResFromOtherNode\":0.0,\"AppProcessMaxTime\":0.0,\"SvrReqFromOtherNode\":0.0,\"TpsThres\":50.0},{\"DateTime\":123456789012,\"ClientReq\":0.0,\"SihProcessTime\":0.0,\"ClientRes\":0.0,\"SvrRes\":0.0,\"ClientResAvMsgLen\":0.0,\"AppProcessTime\":0.0,\"SvrResAvMsgLen\":0.0,\"AppName\":\"TRR\",\"AppResponseMaxTime\":0.0,\"OtherSide\":\"MCSS_COMM01\",\"ClientReqAvMsgLen\":0.0,\"SvrReqAvMsgLen\":0.0,\"SvrReq\":0.0,\"SvrReqTps\":0.0,\"ClientResTps\":0.0,\"CoreProcessTime\":0.0,\"CoreProcessMaxTime\":0.0,\"SihProcessMaxTime\":0.0,\"Discard\":0.0,\"AppResponseTime\":0.0,\"MsgType\":\"No message2\",\"ClientResFromOtherNode\":0.0,\"AppProcessMaxTime\":0.0,\"SvrReqFromOtherNode\":0.0,\"TpsThres\":50.0,\"AllMsg\":\"wee\",\"TransactionID\":\"232\"}";
//	  String demo = "{\"address\":\"北京\",\"name\":\"tiger\",\"age\":44.0}";
	    //	    System.err.println("----->" + cols.length);
//		XContentBuilder mapping = XContentFactory.jsonBuilder()
//				.startObject()
//					.startObject(type)
//						.startObject("properties")
//							.startObject("0").field("type", "string").field("index", "not_analyzed")
//							.endObject()
//						.endObject()
//					.endObject()
//				.endObject();
		
		
		XContentBuilder mapping = XContentFactory.jsonBuilder()
		.startObject()
			.startObject(type)
				.startObject("properties");
		
		Map<String,Object> objectToObj = ConvertUtil.objectToObj(demo, Map.class);
		for(Entry<String, Object> en : objectToObj.entrySet()){
			if(en.getValue() instanceof String){
				mapping.startObject(en.getKey()).field("type", "string").field("index", "not_analyzed").endObject();
			}else if(en.getValue() instanceof Integer){
				mapping.startObject(en.getKey()).field("type", "int").field("index", "not_analyzed").endObject();
			}else if(en.getValue() instanceof Double){
				mapping.startObject(en.getKey()).field("type", "double").field("index", "not_analyzed").endObject();
			}else if(en.getValue() instanceof Long){
				mapping.startObject(en.getKey()).field("type", "long").field("index", "not_analyzed").endObject();
			}
		}
		
		mapping.endObject().endObject().endObject();
		return mapping;
	}
}